﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Microsoft.VisualStudio.TestTools.UnitTesting;

namespace Task_Parallel_Library
{
    [TestClass]
    public class DataFlow
    {
        [TestMethod]
        public void BufferBlock()
        {
            // Create a BufferBlock<int> object. 
            var bufferBlock = new BufferBlock<int>();

            // Post several messages to the block. 
            for (int i = 0; i < 3; i++)
            {
                bufferBlock.Post(i);
            }

            // Receive the messages back from the block. 
            var result = Enumerable.Range(0, 3).Select(x => x == bufferBlock.Receive()).ToList();

            Assert.IsFalse(result.Any(x => x == false));
        }

        [TestMethod]
        public void BroadcastBlock()
        {
            // Create a BroadcastBlock<double> object. 
            var broadcastBlock = new BroadcastBlock<double>(null);

            // Post a message to the block.
            broadcastBlock.Post(Math.E);
            //only last message stays
            broadcastBlock.Post(Math.PI);

            // Receive the messages back from the block several times. 
            for (int i = 0; i < 3; i++)
            {
                Assert.AreEqual(Math.PI, broadcastBlock.Receive());
            }
        }

        [TestMethod]
        public void WriteOnceBlock()
        {
            // Create a WriteOnceBlock<string> object. 
            var writeOnceBlock = new WriteOnceBlock<string>(null);

            // Post several messages to the block in parallel. The first  
            // message to be received is written to the block.  
            // Subsequent messages are discarded.
            Parallel.Invoke(
               () => writeOnceBlock.Post("Message 1"),
               () => writeOnceBlock.Post("Message 2"),
               () => writeOnceBlock.Post("Message 3"));

            // Receive the message from the block.
            Assert.IsNotNull(writeOnceBlock.Receive());
        }

        [TestMethod]
        public void ActionBlock()
        {
            var bag = new ConcurrentBag<int>();
            var actionBlock = new ActionBlock<int>(n => bag.Add(n));

            // Post several messages to the block. 
            for (int i = 0; i < 3; i++)
            {
                actionBlock.Post(i * 10);
            }

            // Set the block to the completed state and wait for all  
            // tasks to finish.
            actionBlock.Complete();
            actionBlock.Completion.Wait();

            Assert.IsTrue(bag.Contains(20));
        }

        [TestMethod]
        public void TransformBlock()
        {
            // Create a TransformBlock<int, double> object that  
            // computes the square root of its input. 
            var transformBlock = new TransformBlock<int, double>(n => Math.Sqrt(n));

            // Post several messages to the block.
            transformBlock.Post(9);
            transformBlock.Post(20);
            transformBlock.Post(30);

            // Read the output messages from the block. 
            Assert.AreEqual(3, transformBlock.Receive());
        }

        [TestMethod]
        public void TransformManyBlock()
        {
            var msg = new[] { "Hello", "World" };

            // Create a TransformManyBlock<string, char> object that splits 
            // a string into its individual characters. 
            var transformManyBlock = new TransformManyBlock<string, char>(s => s.ToCharArray());

            // Post two messages to the first block.
            transformManyBlock.Post(msg[0]);
            transformManyBlock.Post(msg[1]);

            // Receive all output values from the block. 
            var expected = msg.Aggregate("", (total, current) => total + current);
            var result = expected.Select(x => transformManyBlock.Receive()).ToList();
            
            var actual = result.Aggregate("", (total, current) => total + current);
            Assert.AreEqual(expected,actual);
        }

        [TestMethod]
        public void BatchBlock()
        {
            // Create a BatchBlock<int> object that holds ten 
            // elements per batch. 
            var batchBlock = new BatchBlock<int>(10);

            // Post several values to the block. 
            var items = Enumerable.Range(0, 13).ToList();
            items.ForEach(x => batchBlock.Post(x));
            
            // Set the block to the completed state. This causes 
            // the block to propagate out any any remaining 
            // values as a final batch.
            batchBlock.Complete();

            // Print the sum of both batches.

            CollectionAssert.AreEqual(items.Take(10).ToList(), batchBlock.Receive());
            CollectionAssert.AreEqual(items.Skip(10).ToList(), batchBlock.Receive());
        }

        [TestMethod]
        public void JoinBlock()
        {
            // Create a JoinBlock<int, int, char> object that requires 
            // two numbers and an operator. 
            var joinBlock = new JoinBlock<int, int, char>();

            // Post two values to each target of the join.

            joinBlock.Target1.Post(3);
            joinBlock.Target1.Post(6);

            joinBlock.Target2.Post(5);
            joinBlock.Target2.Post(4);

            joinBlock.Target3.Post('+');
            joinBlock.Target3.Post('-');

            // Receive each group of values and apply the operator part 
            // to the number parts. 

            for (int i = 0; i < 2; i++)
            {
                var data = joinBlock.Receive();
                switch (data.Item3)
                {
                    case '+':
                        Console.WriteLine("{0} + {1} = {2}",
                           data.Item1, data.Item2, data.Item1 + data.Item2);
                        break;
                    case '-':
                        Console.WriteLine("{0} - {1} = {2}",
                           data.Item1, data.Item2, data.Item1 - data.Item2);
                        break;
                    default:
                        Console.WriteLine("Unknown operator '{0}'.", data.Item3);
                        break;
                }
            }
        }
    }
}
